package com.avris.tool;

import cn.hutool.core.date.DateUtil;
import cn.hutool.core.io.FileUtil;
import cn.hutool.core.util.StrUtil;
import com.avris.tool.bean.Parameter;
import com.avris.tool.consumer.KafkaConsumerClient;
import com.avris.tool.function.Consumer;
import com.avris.tool.function.ConsumerForTime;
import com.sun.org.apache.xpath.internal.operations.Bool;
import org.apache.kafka.clients.consumer.ConsumerRecord;
import org.apache.kafka.clients.consumer.ConsumerRecords;
import org.apache.kafka.clients.consumer.KafkaConsumer;

import java.io.File;
import java.nio.charset.StandardCharsets;
import java.util.Objects;

import static com.avris.tool.util.ConsumerUtil.*;

/**
 * @author Jast
 * @description
 * @date 2022-09-13 13:54
 */
public class KafkaTool {



    public static void main(String[] args) throws InterruptedException {
        String brokerList = null;
        String topic =  null;
        String groupID =  null;
        Long ts = null;
        Long endTs = null;
        String containsContent = null;
        String type=null;
        Boolean latest = false;

        for (String arg : args) {
            if(arg.startsWith("--")){
                String optionText = arg.substring(2, arg.length());
                String optionValue = null;
                String optionName ;
                if(optionText.contains("=")){
                    optionName = optionText.substring(0,optionText.indexOf("="));
                    optionValue = optionText.substring(optionText.indexOf("=")+1,optionText.length());
                    switch (optionName){
                        case "brokerList": brokerList = optionValue; break;
                        case "topic": topic = optionValue; break;
                        case "group": groupID = optionValue; break;
                        case "beginTime": ts = Long.parseLong(optionValue); break;
                        case "endTime": endTs = Long.parseLong(optionValue); break;
                        case "content": containsContent = optionValue; break;
                        case "type":type = optionValue;break;
                        case "latest":latest = Boolean.parseBoolean(optionValue);break;
                    }
                }
            }
        }
        Parameter parameter = new Parameter();
        parameter.setBrokerList(brokerList);
        parameter.setTopic(topic);
        parameter.setGroup(groupID);
        parameter.setBeginTime(ts);
        parameter.setEndTime(endTs);
        parameter.setContent(containsContent);
        parameter.setType(type);
        parameter.setLatest(latest);

        if(StrUtil.isBlank(type) ){
            System.out.println("启动失败,未指定调用类型type");
            System.exit(1);
        }



        switch (type){
            case "ConsumerForTime":
                ConsumerForTime.run(parameter);
                break;
            case "Consumer":
                Consumer.run(parameter);
                break;
        }



    }



}
